#include <pthread.h>
#include <rid.h>
#include <stdint.h>
#include <stdlib.h>

#define NIT_SHORT_NAMES
#include <nit/macros.h>
#include <nit/palloc.h>
#include <nit/list.h>
#include <nit/set.h>
#include <nit/map.h>

#include "oas.h"

/* We can't make the system do anything, we can only tell it what it should do.
 */

struct Oas_actor {
	Rid id;
	Oas_mail *mail;
	Oas_mail **mail_end;
	pthread_mutex_t mail_mutex;
	void *local;
	Oas_sys *sys;
	Oas_begin begin;
	Oas_act act;
	Oas_die die;
	pthread_t thread;
};


int
oas_init(Oas_sys *sys, void *global, Oas_lost lost)
{
	if (map_init(&sys->actors, 0) < 0)
		return -1;

	if (pthread_rwlock_init(&sys->actors_lock, NULL))
		goto actors;

	if (pthread_mutex_init(&sys->olds_mutex, NULL))
		goto olds;

	if (pthread_mutex_init(&sys->mails_mutex, NULL))
		goto mails;

	if (pthread_cond_init(&sys->done_cond, NULL))
		goto done_cond;

	if (pthread_mutex_init(&sys->done_mutex, NULL))
		goto done_mutex;

	sys->entries = NULL;
	sys->olds = NULL;
	sys->mails = NULL;
	sys->last_set = 0;
	sys->global = global;
	sys->lost = lost;
	return 0;
done_mutex:
	pthread_cond_destroy(&sys->done_cond);
done_cond:
	pthread_mutex_destroy(&sys->mails_mutex);
mails:
	pthread_mutex_destroy(&sys->olds_mutex);
olds:
	pthread_rwlock_destroy(&sys->actors_lock);
actors:
	map_dispose(&sys->actors);
	return -1;
}


static Oas_actor *
get_actor(Oas_sys *sys)
{
	Oas_actor *actor;

	pthread_mutex_lock(&sys->olds_mutex);

	if ((actor = sys->olds))
		sys->olds = actor->local;
	else
		actor = palloc(actor);

	pthread_mutex_unlock(&sys->olds_mutex);
	return actor;
}

static void
recycle_actor(Oas_sys *sys, Oas_actor *actor)
{
	pthread_mutex_lock(&sys->olds_mutex);
	actor->local = sys->olds;
	sys->olds = actor;
	pthread_mutex_unlock(&sys->olds_mutex);
}

static void
broadcast_done(Oas_sys *sys)
{
		pthread_mutex_lock(&sys->done_mutex);
		pthread_cond_broadcast(&sys->done_cond);
		pthread_mutex_unlock(&sys->done_mutex);
}

static void
remove_actor(Oas_sys *sys, Oas_actor *actor,
	     int *done, int *prev, pthread_t *prev_thread)
{
	pthread_rwlock_wrlock(&sys->actors_lock);
	map_remove(&sys->actors, actor->id, sizeof(Rid), &sys->entries);

        if (!sys->actors.entry_num)
		*done = 1;

	if (sys->last_set) {
		*prev = 1;
		*prev_thread = sys->last_thread;
	}

	sys->last_thread = actor->thread;
	sys->last_set = 1;
	pthread_rwlock_unlock(&sys->actors_lock);
}

static void
recycle_mail(Oas_sys *sys, Oas_actor *actor)
{
	pthread_mutex_lock(&sys->mails_mutex);
	*actor->mail_end = sys->mails;
	sys->mails = actor->mail;
	pthread_mutex_unlock(&sys->mails_mutex);
}

static void
return_mail(Oas_sys *sys, Oas_mail *mail)
{
	pthread_mutex_lock(&sys->mails_mutex);
	mail->next = sys->mails;
	sys->mails = mail;
	pthread_mutex_unlock(&sys->mails_mutex);
}

static void *
get_msg(Oas_actor *actor, Rid sender, Oas_sys *sys)
{
	void *msg = NULL;
	Oas_mail *mail;

	pthread_mutex_lock(&actor->mail_mutex);

	if ((mail = actor->mail)) {
		rid_set(sender, mail->sender);
		msg = mail->msg;

		if (!(actor->mail = mail->next))
			actor->mail_end = &actor->mail;
	}

	pthread_mutex_unlock(&actor->mail_mutex);

	if (mail)
		return_mail(sys, mail);

	return msg;
}

static void *
act_routine(void *arg)
{
	Oas_actor *actor = arg;
	Oas_sys *sys = actor->sys;
	int done = 0;
	int prev = 0;
	Rid sender;
	pthread_t prev_thread;

	if (actor->begin && actor->begin(actor->id, actor->local, sys))
		goto done;

        while (actor->act(actor->id, sender, get_msg(actor, sender, sys),
			 actor->local, sys));
done:
	remove_actor(sys, actor, &done, &prev, &prev_thread);

	if (actor->die)
		actor->die(actor->id, actor->mail, actor->local, sys);

	recycle_mail(sys, actor);
	pthread_mutex_destroy(&actor->mail_mutex);
	recycle_actor(sys, actor);

	if (prev)
		pthread_join(prev_thread, NULL);

	if (done)
		broadcast_done(sys);

	pthread_exit(NULL);
}

static int
add_actor(Oas_sys *sys, Oas_actor *actor)
{
	int err = 0;

	pthread_rwlock_wrlock(&sys->actors_lock);

	if (map_add(&sys->actors, actor->id, sizeof(Rid),
		    actor, &sys->entries) < 0)
		goto err_add;

	/* we don't want to start to thread while sys is locked */
	pthread_rwlock_unlock(&sys->actors_lock);

	if (pthread_create(&actor->thread, NULL, act_routine, actor)) {
		/* removal of "dead" actor requires locked sys */
		pthread_rwlock_wrlock(&sys->actors_lock);
		goto err_create;
	}

	goto end;

err_create:
	map_remove(&sys->actors, actor->id, sizeof(Rid), &sys->entries);
err_add:
	pthread_mutex_destroy(&actor->mail_mutex);
	actor->local = sys->olds;
	sys->olds = actor;
	err = -1;
	pthread_rwlock_unlock(&sys->actors_lock);
end:
	return err;
}

int
oas_add(Oas_sys *sys, const Rid id, Oas_mail *mail, void *local,
	Oas_begin begin, Oas_act act, Oas_die die)
{
	Oas_actor *actor = get_actor(sys);

	pcheck(actor, -1);

	if (pthread_mutex_init(&actor->mail_mutex, NULL)) {
		recycle_actor(sys, actor);
		return -1;
	}

	rid_set(actor->id, id);
	actor->mail = mail;
	actor->mail_end = &actor->mail;
	actor->local = local;
	actor->sys = sys;
	actor->begin = begin;
	actor->act = act;
	actor->die = die;
	return add_actor(sys, actor);
}

static Oas_mail *
get_mail(Oas_sys *sys)
{
	Oas_mail *mail = NULL;

	pthread_mutex_lock(&sys->mails_mutex);

	if ((mail = sys->mails))
		sys->mails = mail->next;
	else
		mail = palloc(mail);

	pthread_mutex_unlock(&sys->mails_mutex);
	return mail;
}

static void
send_mail(Oas_actor *actor, Oas_mail *mail)
{
	pthread_mutex_lock(&actor->mail_mutex);
	*actor->mail_end = mail;
	actor->mail_end = &mail->next;
	pthread_mutex_unlock(&actor->mail_mutex);
}

int
oas_msg(Oas_sys *sys, const Rid sender, const Rid to, void *msg)
{
	Oas_actor *actor;
	Oas_mail *mail;

	if (!(mail = get_mail(sys)))
		return -1;

	mail->msg = msg;
	rid_set(mail->sender, sender);
	mail->next = NULL;

	pthread_rwlock_rdlock(&sys->actors_lock);

	if (!(actor = map_get(&sys->actors, to, sizeof(Rid)))) {
		pthread_rwlock_unlock(&sys->actors_lock);
		return_mail(sys, mail);
		return sys->lost(sender, msg, sys);
	}

	send_mail(actor, mail);
	pthread_rwlock_unlock(&sys->actors_lock);
	return 0;
}

void
oas_wait(Oas_sys *sys)
{
	/* entry_num cannot be set to zero while locking actors_mutex. */
	pthread_rwlock_rdlock(&sys->actors_lock);

	/* It shouldn't matter if all actors finish or not before the loop. */
	while (sys->actors.entry_num) {
		pthread_mutex_lock(&sys->done_mutex);
		pthread_rwlock_unlock(&sys->actors_lock);
		pthread_cond_wait(&sys->done_cond, &sys->done_mutex);
		pthread_mutex_unlock(&sys->done_mutex);
		pthread_rwlock_rdlock(&sys->actors_lock);
	}

	pthread_rwlock_unlock(&sys->actors_lock);

	if (sys->last_set)
		pthread_join(sys->last_thread, NULL);
}

static void
free_olds(Oas_actor *olds)
{
	while (olds) {
		Oas_actor *next = olds->local;

		free(olds);
		olds = next;
	}
}

static void
free_mail(Oas_mail *mail)
{
	while (mail) {
		Oas_mail *next = mail->next;

		free(mail);
		mail = next;
	}
}

void *
oas_dispose(Oas_sys *sys)
{
	map_dispose(&sys->actors);
	entries_free(sys->entries);
	pthread_rwlock_destroy(&sys->actors_lock);
	free_olds(sys->olds);
	pthread_mutex_destroy(&sys->olds_mutex);
	free_mail(sys->mails);
	pthread_mutex_destroy(&sys->mails_mutex);
	pthread_cond_destroy(&sys->done_cond);
	pthread_mutex_destroy(&sys->done_mutex);
	return sys->global;
}
